package org.niit.service

import java.text.SimpleDateFormat

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.niit.common.TService
import org.niit.dao.RealTimeDao
import org.niit.handler.DataHandler
import org.niit.bean.AdClickData
import org.niit.util.MyKafkaUtil

class RealTimeService extends TService{

  override def dataAnalysis(data:DStream[AdClickData]): Any = {


    reduceAdCount(data)


  }
  //统计每天 每地区 每城市 的广告点击次数
  private def reduceAdCount(adClickData: DStream[AdClickData]): Unit ={
    /*
    1.
    在 Kafka获得的数据     ==> 时间戳 地区 城市 广告 用户
    提取数据作为合并的键     ==>  时间戳  地区  城市 广告
                Key      ==>  日期（2023-4-4 15:03:07）    地区(华东)  城市（上海）  广告（士力架）
                value    ==>    1
      2.
      最终格式 （ ( 日期,地区,城市,广告),1） （ ( 日期,地区,城市,广告),1） （ ( 日期,地区,城市,广告),1）
                （ ( 日期,地区,城市,广告),3）  存入数据库
     */
    val reduceDS =  adClickData.map( data=> {
      val sdf = new SimpleDateFormat("yyyy-MM-dd")
      val day = sdf.format(new java.util.Date(data.ts.toLong))
      val area = data.area
      val city = data.city
      val ad = data.ad
      // （ ( 日期,地区,城市,广告),1）
      ( (day,area,city,ad),1 )
    })
    //根据联合主键进行合并 （ ( 日期,地区,城市,广告),1） （ ( 日期,地区,城市,广告),1） （ ( 日期,地区,城市,广告),1）==》（ ( 日期,地区,城市,广告),3）
    val resData =  reduceDS.reduceByKey(_+_)
    //存入数据库

    resData.foreachRDD(rdd=>{
      rdd.foreach{
        case ( (day,area,city,ad),count )=>{
          println(s"${day},${area},${city},${ad},${count}")
          val realTimeDao = new RealTimeDao
          realTimeDao.insertRealTimeAdCount(day,area,city,ad,count)
        }
      }
    })


  }


}
